package cn.itcast.tags.models.ml

import cn.itcast.tags.models.{AbstractModel, ModelType}
import cn.itcast.tags.tools.TagTools
import org.apache.spark.broadcast.Broadcast
import org.apache.spark.ml.classification.{DecisionTreeClassificationModel, DecisionTreeClassifier}
import org.apache.spark.ml.evaluation.MulticlassClassificationEvaluator
import org.apache.spark.ml.feature.{VectorAssembler, VectorIndexer, VectorIndexerModel}
import org.apache.spark.sql.expressions.UserDefinedFunction
import org.apache.spark.sql.{Column, DataFrame, SparkSession}
import org.apache.spark.sql.functions._

/**
 * 用户购物性别
 */
class UsgModel extends AbstractModel("用户购物性别USF",ModelType.ML){
  override def doTag(businessDF: DataFrame, tagDF: DataFrame): DataFrame = {
    val session: SparkSession = businessDF.sparkSession
    import session.implicits._
    /*root
          |-- cordersn: string (nullable = true)
          |-- ogcolor: string (nullable = true)
          |-- producttype: string (nullable = true)*/

    // 1. 获取订单表数据tbl_orders，与订单商品表数据关联获取会员ID
    val ordersDF: DataFrame = spark.read
      .format("hbase")
      .option("zkHosts", "bigdata-cdh01.itcast.cn")
      .option("zkPort", "2181")
      .option("hbaseTable", "tbl_tag_orders")
      .option("family", "detail")
      .option("selectFields", "memberid,ordersn")
      .load()

    ordersDF.show(10,false)

    // 2. 加载维度表数据：tbl_dim_colors（颜色）、tbl_dim_products（产品）
    // 2.1 加载颜色维度表数据
    val colorsDF: DataFrame = {
      spark.read
        .format("jdbc")
        .option("driver", "com.mysql.jdbc.Driver")
        .option("url", "jdbc:mysql://bigdata-cdh01.itcast.cn:3306/?useUnicode=true&characterEncoding=UTF8&serverTimezone=UTC")
        .option("dbtable", "profile_tags.tbl_dim_colors")
        .option("user", "root")
        .option("password", "123456")
        .load()
    }
    colorsDF.show(10,false)
    // 2.2. 构建颜色WHEN语句
    val colorColumn: Column = {
      // 声明变量
      var colorCol: Column = null
      colorsDF
        .as[(Int, String)].rdd
        .collectAsMap()
        .foreach { case (colorId, colorName) =>
          if (null == colorCol) {
            colorCol = when($"ogcolor".equalTo(colorName), colorId)
          } else {
            colorCol = colorCol.when($"ogcolor".equalTo(colorName),colorId)
          }
        }
      colorCol = colorCol.otherwise(0).as("color")
      // 返回
      colorCol
    }

    // 2.3 加载商品维度表数据
    val productsDF: DataFrame = {
      spark.read
        .format("jdbc")
        .option("driver", "com.mysql.jdbc.Driver")
        .option("url",
          "jdbc:mysql://bigdata-cdh01.itcast.cn:3306/?useUnicode=true&characterEncoding=UTF8&serverTimezone=UTC")
        .option("dbtable", "profile_tags.tbl_dim_products")
        .option("user", "root")
        .option("password", "123456")
        .load()
    }
    productsDF.show(10,false)

    // 2.4. 构建颜色WHEN语句
    var productColumn: Column = {
      // 声明变量
      var productCol: Column = null
      productsDF
        .as[(Int, String)].rdd
        .collectAsMap()
        .foreach { case (productId, prodcutName) =>
          if (null == productCol) {
            productCol = when($"producttype".equalTo(prodcutName), productId)
          } else {
            productCol =
              productCol.when(col("producttype").equalTo(prodcutName), productId)
          }
        }
      productCol = productCol.otherwise(0).as("product")
      productCol
    }

    // 根据运营规则标注的部分数据
    val labelColumn: Column = {
      when($"ogcolor".equalTo("樱花粉")
        .or($"ogcolor".equalTo("粉色"))
        .or($"ogcolor".equalTo("白色"))
        .or($"ogcolor".equalTo("香槟色"))
        .or($"ogcolor".equalTo("香槟金"))
        .or($"productType".equalTo("料理机"))
        .or($"productType".equalTo("挂烫机"))
        .or($"productType".equalTo("吸尘器/除螨仪")), 1) //女
        .otherwise(0)//男
        .alias("label")//决策树预测label
    }

    //关联orders数据，颜色维度和商品类别维度
    val goodsDF: DataFrame = businessDF
      .join(ordersDF, businessDF("cordersn") === ordersDF("ordersn"))
      .select(
        $"memberid".as("userId"),
        colorColumn, //颜色
        productColumn, //产品类别
        labelColumn
      )

    goodsDF.printSchema()
    /*
           |-- userId: string (nullable = true)
           |-- color: integer (nullable = false)
           |-- product: integer (nullable = false)
           |-- label: integer (nullable = false)
     */

    //直接使用标注数据，给用户打标签
    val predictionDF: DataFrame = goodsDF.select($"userId", $"label".as("prediction"))

    predictionDF.show(100,false)
    //按照用户ID进行分组，统计每个用户购物男性或女性个数及占比
    val genderDF: DataFrame = predictionDF
      .groupBy($"userId")
      .agg(
        count($"userId").as("total"),
        //判断label为0时，表示男性商品，设置为1，使用sum累加
        sum(
          when($"prediction".equalTo(0), 1).otherwise(0)
        ).as("maleTotal"),
        sum(
          when($"prediction".equalTo(1), 1).otherwise(0)
        ).as("femaleTotal")
      )
      //|userId|total|maleTotal|femaleTotal|

    //计算标签，
    //获取属性标签
    val rulesMap: Map[String, String] = TagTools.convertMap(tagDF)
    val rulesBroadcast: Broadcast[Map[String, String]] = session.sparkContext.broadcast(rulesMap)
    // 对每个用户，分别计算男性商品和女性商品占比，当占比大于0.6时，确定购物性别
    val gender_tag_udf: UserDefinedFunction = udf(
      (total: Long, maleTotal: Long, femaleTotal: Long) => {
        val maleRate: Double = maleTotal / total.toDouble
        val femaleRate: Double = femaleTotal / total.toDouble
        if (maleRate >= 0.6) {
          rulesBroadcast.value("0")
        } else if (femaleRate >= 0.6) {
          rulesBroadcast.value("1")
        } else {
          rulesBroadcast.value("-1")
        }
      }
    )

    //获取画像标签数据
    val modelDF: DataFrame = genderDF.select(
      $"userId",
      gender_tag_udf($"total", $"maleTotal", $"femaleTotal").as("usg")
    )

    modelDF.show(100,false)
    null
  }
}

object UsgModel{
  def main(args: Array[String]): Unit = {
    val model = new UsgModel
    model.executeModel(378L,false)
  }
}